package com.spx.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * create by undeRdoG on  2022-04-26  23:28
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class BoundedStreamWordCount {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //env.setParallelism(1);


        //DataStreamSource<String> source = env.readTextFile("input/word.txt");

        DataStreamSource<String> source = env.socketTextStream("hadoop102", 7777);


        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = source.flatMap((String line, Collector<Tuple2<String,Long>>collector) -> {
            String[] fields = line.split(" ");
            for (String filed : fields) {
                collector.collect(Tuple2.of(filed, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING,Types.LONG));


        KeyedStream<Tuple2<String, Long>, String> keyedStream = wordAndOne.keyBy(
                 stringLongTuple2 -> stringLongTuple2.f0

        );


        SingleOutputStreamOperator<Tuple2<String, Long>> res = keyedStream.sum(1);

        res.print();

        env.execute();
    }
}
